package ar.edu.itba.pod.Legajo47342.communication;

import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import ar.edu.itba.pod.Legajo47342.Node;
import ar.edu.itba.pod.Legajo47342.communication.payload.ResourceTransferMessagePayloadImpl;
import ar.edu.itba.pod.simul.communication.ThreePhaseCommit;
import ar.edu.itba.pod.simul.communication.Transactionable;
import ar.edu.itba.pod.simul.communication.payload.Payload;
import ar.edu.itba.pod.simul.market.Resource;

public class TransactionableImpl implements Transactionable {
	
	private AtomicBoolean inTrans;
	//	private AtomicBoolean waiting;
	private Node node;
	private String remoteNodeId;
	private AtomicBoolean doneExchange;
	private Resource res;
	private AtomicInteger amount;

	public TransactionableImpl(Node node) throws RemoteException {
		super();
		this.inTrans = new AtomicBoolean(false);
		//this.waiting.set(false);
		this.doneExchange = new AtomicBoolean(false);
		this.amount = new AtomicInteger(0);
		this.node = node;
		
		UnicastRemoteObject.exportObject(this, 0);
	}

	@Override
	public void beginTransaction(String remoteNodeId, long timeout)
			throws RemoteException {
		final long timeoutF = timeout;
		
		if (inTrans.get()) {
			throw new IllegalStateException("Already in a trnsaction. Cannot create a new one");
		}
		
		Transactionable remoteTrans = node.getManager().
									getConnectionManager(remoteNodeId).getNodeCommunication();
		
		//start waiting here
		Thread waitThread = new Thread(new Runnable() {
			
			@Override
			public void run() {
				//waiting.set(true);
				try {
					Thread.sleep(timeoutF);
				} catch (InterruptedException e) {
					System.out.println("Wait interrupted, continue setting transaction");
					//waiting.set(false);
					return;
				}
				//aca se paso el timeout y tengo que anular la creacion de la transaction
				throw new IllegalStateException("Timeout reached. Canceled creation of transaction");
			}
		});
		waitThread.setDaemon(true);
		waitThread.start();
		
		try {
			System.out.println("POR ENTRAR A ACCEPT");
			remoteTrans.acceptTransaction(node.getNodeId());
			//retorne antes del timeout. termino el thread que espera y sigo seteando la transaccion
			waitThread.interrupt();
			this.inTrans.set(true);
			this.remoteNodeId = remoteNodeId;
		} catch (Exception e) {
			e.printStackTrace();
			//couldn't connect
			return;
		}
		
	}

	@Override
	public void acceptTransaction(String remoteNodeId) throws RemoteException {
		if (inTrans.get()) {
			try {
				Thread.sleep(Node.TRANS_WAIT);//wait logout here
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
			
			if (inTrans.get()) {
				throw new IllegalStateException("The node is still in a transaction");
			}
		}
		synchronized (inTrans) {
			this.inTrans.set(true);
		}
		this.remoteNodeId = remoteNodeId;
	}

	@Override
	public void endTransaction() throws RemoteException {
		if (!inTrans.get()) {
			throw new IllegalStateException("The node wasn't in a transaction");
		}
		
		threePhaseCommit();
		
		synchronized (inTrans) {
			inTrans.set(false);
		}
		remoteNodeId = null;
		
	}

	private void threePhaseCommit() {
		try {
			ThreePhaseCommit local3PC = node.getManager().getThreePhaseCommit();
			ThreePhaseCommit remote3PC = node.getManager().getConnectionManager(remoteNodeId).getThreePhaseCommit();
			
			boolean localResp = local3PC.canCommit(node.getNodeId(), Node.TPCWAIT);
			boolean remoteResp = remote3PC.canCommit(node.getNodeId(), Node.TPCWAIT);
			
			if (!remoteResp || !localResp) {
				return;
			}
			
			try {
				local3PC.preCommit(node.getNodeId());
				remote3PC.preCommit(node.getNodeId());
				
				local3PC.doCommit(node.getNodeId());
				remote3PC.doCommit(node.getNodeId());
			} catch (RemoteException e) {
				e.printStackTrace();
				local3PC.abort();
				remote3PC.abort();
			}
			
		} catch (RemoteException e) {
			e.printStackTrace();
		}
		
	}

	@Override
	public void exchange(Resource resource, int amount, String sourceNode,
			String destinationNode) throws RemoteException {
		if (!inTrans.get()) {
			throw new IllegalStateException("The node wasn't in a transaction");
		}
		if (doneExchange.get()) {
			throw new IllegalStateException("Cannot exchange twice in the same transaction context");
		}
		if (!(node.getNodeId().equals(sourceNode) || node.getNodeId().equals(destinationNode)) || 
				!(remoteNodeId.equals(sourceNode) || remoteNodeId.equals(destinationNode))
				|| sourceNode.equals(destinationNode)) {
			throw new IllegalStateException("Error in exchange nodes ids");
		}
		
		this.res = resource;
		this.amount.set(amount);
		this.doneExchange.set(true);
	}

	@Override
	public Payload getPayload() throws RemoteException {
		if (!inTrans.get()) {
			throw new IllegalStateException("Not in a transaction. Cannot return the payload");
		}
		if (!doneExchange.get()) {
			throw new IllegalStateException("Exchange is not done. Cannot return payload.");
		}
		
		return new ResourceTransferMessagePayloadImpl(res, amount.get(), node.getNodeId(), remoteNodeId);
	}

	@Override
	public void rollback() throws RemoteException {
		synchronized (doneExchange) {
			doneExchange.set(false);
			res = null;
			amount.set(0);
		}
		
		synchronized (inTrans) {
			inTrans.set(false);
		}
	}

}
